/*
    This file is part of FSJController.

    FSJController is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    FSJController is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with FSJController.  If not, see <http://www.gnu.org/licenses/>.
*/

package de.qsc.fsjcontroller;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of a Freeswitch controller instance that connects to a remove Freeswitch server on
 * the event socket interface
 * 
 * @author Roland Hänel <rha@qsc.de>
 */
public class FSController implements Runnable {

    private Executor executor ;
    
    private static final Logger logger = LoggerFactory.getLogger(FSController.class);

    private static final int MESSAGE_BUFFER_SIZE = 200000;

    private String host;
    private int port;
    private String password;

    private SocketChannel sock;
    private Selector sel;
    private ByteBuffer readbuf;
    private ByteBuffer writebuf;

    private  String header;
    private String body;
    private int expectedBodySize;
    private boolean connected;
    
    private FSChannelProvider channelProvider;

    private enum AuthenticationState {
        INIT, AUTH_TRYING, EVENT_INIT, OK, FAILED
    };

    private AuthenticationState authState;

    private BlockingQueue<FSCommand> queuedCommands;
    private Map<UUID, FSAPICommand> activeCommands;

    private Map<String, FSConferenceHandler> conferences;

    private Set<FSEventCallback> eventCallbacks;
    
    private Map<UUID, FSChannel> channels;
    private Map<UUID, Queue<FSEvent>> channelEventQueues;

    private static final Pattern contentLengthPattern = Pattern.compile("^\\s*Content-Length\\s*:\\s*(\\d+)\\s*$", Pattern.CASE_INSENSITIVE
            | Pattern.MULTILINE);
    private static final Pattern contentTypePattern = Pattern.compile("^\\s*Content-Type\\s*:\\s*(\\S+)\\s*$", Pattern.CASE_INSENSITIVE
            | Pattern.MULTILINE);
    private static final Pattern replyTextPattern = Pattern.compile("^\\s*Reply-Text\\s*:\\s*([\\S ]+)\\s*$", Pattern.CASE_INSENSITIVE
            | Pattern.MULTILINE);
    private static final Pattern jobUUIDPattern = Pattern.compile(
            "^\\s*Job-UUID\\s*:\\s*([a-f0-9]{8}\\-[a-f0-9]{4}\\-[a-f0-9]{4}\\-[a-f0-9]{4}\\-[a-f0-9]{12})\\s*$",
            Pattern.CASE_INSENSITIVE | Pattern.MULTILINE);
    private static final Pattern commandErrPattern = Pattern.compile(
            "^\\s*Content-Length\\s*:\\s*(\\d+)\\s*$.*^$^.*(-ERR)",
            Pattern.CASE_INSENSITIVE | Pattern.MULTILINE | Pattern.DOTALL);
    
    private boolean inShutdown = false;
    
    /**
     * Creates a Freeswitch Controller instance
     * 
     * @param host
     *            hostname of the Freeswitch server
     * @param port
     *            port number of the Freeswitch server event socket listener
     * @param password
     *            password of the event socket listener interface
     * @param channelProvider
     *            the channel provider that creates FSChannel objects for incoming calls
     * @param executor
     *            an executor interface used to run callbacks out of the FSController's I/O thread  
     */
    public FSController(String host, int port, String password, final FSChannelProvider channelProvider, Executor executor) {
        
        this.host = host;
        this.port = port;
        this.password = password;
        this.executor = executor;
        this.channelProvider = channelProvider;
        
        readbuf = ByteBuffer.allocate(MESSAGE_BUFFER_SIZE);
        writebuf = ByteBuffer.allocate(MESSAGE_BUFFER_SIZE);

        connected = false;
        authState = AuthenticationState.INIT;

        queuedCommands = new LinkedBlockingQueue<FSCommand>();
        activeCommands = new HashMap<UUID, FSAPICommand>();
        channels = Collections.synchronizedMap(new HashMap<UUID, FSChannel>());
        channelEventQueues = new HashMap<UUID, Queue<FSEvent>>();
        conferences = Collections.synchronizedMap(new HashMap<String, FSConferenceHandler>());
        eventCallbacks = new HashSet<FSEventCallback>();
        
    }

    /**
     * @return true if the controller is connected to a Freeswitch server
     */
    public boolean isConnected() {
        return connected;
    }

    /**
     * @return true, if the controller is successfully authenticated on a Freeswitch server
     */
    public AuthenticationState isAuthenticated() {
        return authState;
    }

    /**
     * Connects to the remote Freeswitch event socket listener
     */
    public void connect() throws IOException {
        // just to be sure, re-open socket/selector if they are closed
        if (sock == null || !sock.isOpen())
            sock = SocketChannel.open();
        if (sel == null || !sel.isOpen())
            sel = Selector.open();

        // connect to server
        sock.connect(new InetSocketAddress(host, port));
        sock.configureBlocking(false);

        // and create our selector, register socket with it
        sock.register(sel, SelectionKey.OP_READ);

        logger.info("connected to event socket on {}:{}", host, port);

        // reset buffers, queue authentication
        readbuf.clear();
        synchronized (writebuf) {
            writebuf.clear();
        }
        
        connected = true;
        authState = AuthenticationState.INIT;
        header = null;
        body = null;
        expectedBodySize = 0;

        queuedCommands.clear();
        activeCommands.clear();
    }

    /**
     * Disconnects from the remote Freeswitch event socket listener
     */
    private void disconnect() throws IOException {
        logger.info("disconnected from event socket on {}:{}", host, port);
        
        // just to be sure, close everything
        if (sock.isOpen())
            sock.close();
        if (sel.isOpen())
            sel.close();

        // reset buffers, queue authentication
        readbuf.clear();
        synchronized (writebuf) {
            writebuf.clear();
        }

        connected = false;
        authState = AuthenticationState.INIT;
    }
    
    /**
     * Signals the runner thread to terminate and disconnects the socket listener
     * 
     * @throws IOException
     */
    public void shutdown() {
        inShutdown = true;
        try {
            disconnect();
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * Read a header from the read buffer if possible
     */
    public void readHeader() {
        boolean firstNl = false;
        while (readbuf.hasRemaining()) {
            // as long as there's something in the readbuffer, look for NL
            if (readbuf.get() == '\n') {
                // we have a newline
                if (firstNl == true) {
                    // this is the second NL, end-of-header, we have it!
                    int pos = readbuf.position();
                    readbuf.position(0);
                    byte[] headerBuf = new byte[pos];
                    readbuf.get(headerBuf, 0, pos);
                    header = new String(headerBuf);
                    readbuf.compact();
                    readbuf.flip();

                    // look for Content-Length
                    Matcher m = contentLengthPattern.matcher(header);
                    if (m.find()) {
                        // we expect a following body with the specified length
                        expectedBodySize = Integer.parseInt(m.group(1));
                    } else {
                        // no body will follow
                        expectedBodySize = 0;
                    }
                    return;
                } else {
                    // this was the first NL, remember that
                    firstNl = true;
                }
            } else {
                // no NL, remember that
                firstNl = false;
            }
        }

    }

    /**
     * Look for message headers/bodies within the read buffer
     */
    public void scanReadBuffer() {
        // flip readbuf to be able to scan through it
        readbuf.flip();
        while (readbuf.hasRemaining()) {
            if (header == null) {
                // we don't have a header, so try to parse one
                readHeader();
                if (header == null) {
                    // didn't get a header, so there's nothing useful in the buffer for now
                    break;
                }
            }

            // here, we know we have a header, look for a body if we expect one
            if (expectedBodySize > 0) {
                if (readbuf.remaining() >= expectedBodySize) {
                    // we're read to read a body
                    byte[] bodyBuf = new byte[expectedBodySize];
                    readbuf.get(bodyBuf, 0, expectedBodySize);
                    body = new String(bodyBuf);
                    readbuf.compact();
                    readbuf.flip();
                } else {
                    // didn't get a body, need to wait for more input
                    break;
                }
            }

            // here we have a complete header + body
            processMessage();
        }
        // set position to end of readbuf to read more bytes
        readbuf.position(readbuf.limit());
        readbuf.limit(readbuf.capacity());
        
        if (readbuf.remaining() == 0) {
            // we have just tried to read something out of the buffer, and it's still full:
            // this signals an error condition, clear the buffer.
            logger.error("read buffer full and no message could be parsed - discard");
            readbuf.clear();            
        }
    }

    /**
     * Processes a message (header + body) that was received on the connection
     */
    public void processMessage() {
        // extract Content-Type and Reply-Text header
        String contentType = "";
        String replyText = "";
        Matcher m;
        m = contentTypePattern.matcher(header);
        if (m.find()) {
            contentType = m.group(1);
        }
        m = replyTextPattern.matcher(header);
        if (m.find()) {
            replyText = m.group(1);
        }

        // if it's a command reply, we extract the UUID and update the command object
        if (authState.equals(AuthenticationState.OK) && contentType.equals("command/reply")) {
            final FSCommand command = queuedCommands.poll();
            
            if (command != null) {
                if (replyText.startsWith("+OK")) {                    
                    if (command instanceof FSAPICommand) {
                        m = jobUUIDPattern.matcher(header);
                        if (m.find()) {
                            UUID uuid = UUID.fromString(m.group(1));
                            activeCommands.put(uuid, (FSAPICommand) command);
                            logger.trace("command {} is now queued with job UUID {}", command, uuid);
                        } else {
                            logger.warn("cannot queue command {}: no job UUID", command);
                        }
                    }
                } else {
                    logger.debug("command {} failed: {}", command, replyText);
                    if (command instanceof FSAPICommand) {
                        final String failText = replyText;
                        final FSAPICommandCallback callback = ((FSAPICommand) command).getCallback();
                        if (callback != null) {
                            executor.execute(new Runnable() {
                                public void run() {
                                    callback.onFailure(failText);
                                }
                            });
                        }
                    }
                }
            } else {
                logger.warn("received command/reply and no command in queue");
            }
        }

        // if it's an event, we create an FSEvent object
        else if (contentType.equals("text/event-plain") && body != null) {

            final FSEvent event = new FSEvent(body);
            UUID channelUUID = event.getChannelUUID();
            logger.trace("received event {}", event);
            
            switch (event.name) {
                case BACKGROUND_JOB :
                    // extract the state for the corresponding FSCommand object
                    
                    String jobUUID = event.getAttribute("Job-UUID");
                    if (jobUUID == null) {
                        logger.warn("event {} does not contain Job-UUID");
                        break;
                    }
                    UUID uuid;
                    try {
                        uuid = UUID.fromString(jobUUID);
                    }
                    catch (IllegalArgumentException e) {
                        logger.warn("cannot parse Job-UUID {} in event {}", jobUUID, event);
                        break;
                    }
                    
                    // remove extracted command from list of active commands
                    final FSAPICommand command = activeCommands.remove(uuid);
                    if (command == null) {
                        logger.warn("did not find active command UUID {}", uuid);
                        break;
                    }
                    
                    logger.trace("API command {} completed with job UUID {}", command, jobUUID);
                    
                    if (!commandErrPattern.matcher(body).find()) {
                        if (command.getCallback() != null) {
                            executor.execute(new Runnable() {
                                public void run() {
                                    command.getCallback().onSuccess(event);
                                }
                                
                            });
                        }
                    } else {
                        if (command.getCallback() != null) {
                            executor.execute(new Runnable() {
                                public void run() {
                                    command.getCallback().onError(event);
                                }
                            });
                        }
                    }
                    break;
                    
                case CUSTOM_CONFERENCE_MAINTENANCE :
                    // conference related event
                    String conferenceName = event.getAttribute("Conference-Name");
                    if (conferenceName == null) {
                        logger.warn("event {} doesn't have attribute Conference-Name", event);
                        break;
                    }
                    final FSConferenceHandler conference = conferences.get(conferenceName);
                    // if we have that conference, dispatch the event to it
                    if (conference != null) {
                        executor.execute(new Runnable() {
                            public void run() {
                                conference.processEvent(event);
                            }
                        });
                    } else {
                        logger.info("conference {} has no handler - event ignored", conferenceName);
                    }
                    break;
                
                case CHANNEL_CREATE:
                    String direction = event.getAttribute("Call-Direction");
                    if (channelUUID != null && direction != null && direction.equals("inbound")) {
                        logger.trace("event {} triggers new channel UUID {}", event, channelUUID);
                        
                        final UUID cUUID = channelUUID;
                        final Queue<FSEvent> queue = new LinkedList<FSEvent>();

                        // create a queue to buffer events for this channel until it is actually created
                        synchronized (channelEventQueues) {
                            channelEventQueues.put(cUUID, queue);
                        }

                        executor.execute(new Runnable() {
                            public void run() {
                                // create the channel handler object and activate it
                                final FSChannel channel = channelProvider.createChannel(FSController.this, event);
                                if (channel == null) {
                                    logger.trace("channelProvider didn't provide a channel for UUID {}", cUUID);
                                    synchronized (channelEventQueues) {
                                        channelEventQueues.remove(cUUID);
                                    }
                                    return;
                                }
                                channel.setState(FSChannel.State.ACTIVE);
                                logger.trace("channelProvider created new channel UUID {}", cUUID);
                                // now, replay any buffered events to the channel, and then insert the channel
                                // to the channel list so that it can receive further events
                                synchronized (channelEventQueues) {
                                    channelEventQueues.remove(cUUID);
                                    channels.put(cUUID, channel);
                                }
                                for (FSEvent e : queue) {
                                    channel.onEvent(e);
                                }
                            }
                        });
                    }
                    break;

                case CHANNEL_UUID:
                    if (channelUUID != null) {
                        final FSChannel channel = channels.get(channelUUID);
                        if (channel != null) {
                            logger.trace("event {} triggers channel UUID {} to get active", event, channelUUID);
                            channel.setState(FSChannel.State.ACTIVE);
                        } else {
                            logger.trace("event {} cannot find new UUID {}", event, channelUUID);
                        }
                    }
                    break;

                case CHANNEL_DESTROY:
                    if (channelUUID != null) {
                        final FSChannel channel = channels.remove(channelUUID);
                        if (channel != null) {
                            logger.trace("event {} destroys channel UUID {}", event, channelUUID);
                            channel.setState(FSChannel.State.DEAD);
                            executor.execute(new Runnable() {
                                public void run() {
                                    channel.onEvent(event);
                                }
                            });
                        } else {
                            logger.trace("event {} not handling channel for UUID {}", event, channelUUID);
                        }
                    } else {
                        logger.warn("event {} without channel UUID", event);
                    }
                    break;
            }

            // dispatch event into channel or buffer queue
            if (channelUUID != null) {
                synchronized (channelEventQueues) {
                    final FSChannel channel = channels.get(channelUUID);
                    if (channel != null) {
                        executor.execute(new Runnable() {
                            public void run() {
                                channel.onEvent(event);
                            }
                        });
                    } else {
                        // there is no channel object, but there may be a buffer queue for 
                        // a channel object that is in the process of being created
                        Queue<FSEvent> queue = channelEventQueues.get(channelUUID);
                        if (queue != null) {
                            queue.offer(event);
                        }
                    }
                }
            }
                        
            // see if we have a generic handler
            synchronized (eventCallbacks) {
                for (final FSEventCallback callback : eventCallbacks) {
                    if (callback.isApplicable(event)) {
                        executor.execute(new Runnable() {
                            public void run() {
                                callback.handleEvent(event);
                            }
                        });
                    }
                }
            }

        }
        // if the remote asks us to authenticate, we do so
        else if (contentType.equals("auth/request")) {
            synchronized (writebuf) {
                writebuf.put("auth ".getBytes()).put(password.getBytes()).put("\n\n".getBytes());
            }
            authState = AuthenticationState.AUTH_TRYING;
        }

        // if we're not authenticated and get a reply, that's the authentication reply
        else if (authState == AuthenticationState.AUTH_TRYING && contentType.equals("command/reply")) {
            if (replyText.startsWith("+OK")) {
                logger.trace("authentication ok");
                // initialize event stream
                synchronized (writebuf) {
                    /*
                    writebuf.put(new String("event plain " + FSEvent.Name.BACKGROUND_JOB + " " + FSEvent.Name.CHANNEL_CREATE + " " + FSEvent.Name.CHANNEL_UUID + " " + FSEvent.Name.CHANNEL_DESTROY + " "
                            + FSEvent.Name.CHANNEL_EXECUTE + " " + FSEvent.Name.CHANNEL_PARK + " " + FSEvent.Name.CHANNEL_UUID  + " " + FSEvent.Name.CHANNEL_EXECUTE_COMPLETE + " "
                            + "CUSTOM conference::maintenance\n\n").getBytes());
                    */
                    writebuf.put(new String("event plain all\n\n").getBytes());

                }
                authState = AuthenticationState.EVENT_INIT;
            } else {
                logger.warn("authentication failed: {}", replyText);
                authState = AuthenticationState.FAILED;
            }
        }

        // if we've sent the event command and get a reply, that's the event reply
        else if (authState == AuthenticationState.EVENT_INIT && contentType.equals("command/reply")) {
            if (replyText.startsWith("+OK")) {
                logger.trace("event stream initialized");
                authState = AuthenticationState.OK;
            } else {
                authState = AuthenticationState.FAILED;
                logger.warn("event system initialization error: {}", replyText);
            }
        }

        // this means we have unknown/unexpected stuff. see what this is
        else {
            logger.error("unknown event, header {}, body {}", header, body != null ? body : "null");
        }

        header = null;
        body = null;
        expectedBodySize = 0;
    }

    /**
     * Queues a Freeswitch API command for execution on the remote server
     * 
     * @param command
     *            the Freeswitch command to queue for execution
     * @return true if the command was successfully written to the queue
     */
    public boolean queueCommand(FSCommand command) {
        // if we're not authenticated, no chance to issue a command
        if (authState != AuthenticationState.OK) {
            logger.warn("command {} to be queued but controller in auth state {}", command, authState);
            return false;
        }
        synchronized (writebuf) {
            // memorize the command as 'in progress'
            queuedCommands.add(command);
            // write the command to our queue
            writebuf.put(command.getBytes());
        }
        // wake up selector so that our buffer can be written
        if (sel != null) {
            sel.wakeup();
        }
        logger.trace("command {} queued", command);
        return true;
    }

    /**
     * Starts the FSController up
     */
    public void run() {
        while (inShutdown == false) {

            try {
                // connect if we're not already connected
                if (sock == null || !sock.isConnected()) {
                    connect();
                }
                
                // wait for pending I/O
                sel.selectedKeys().clear();
                SelectionKey key = sock.keyFor(sel);
                
                int ops = 0;
                if (readbuf.remaining() > 0) {
                    ops |= SelectionKey.OP_READ;
                }
                if (writebuf.position() > 0) {
                    ops |= SelectionKey.OP_WRITE;
                }
                key.interestOps(ops);
                
                if (sel.select(60000) == 0) {
                    if (writebuf.position() == 0 && sock != null && !sock.isConnected()) {
                        // there is nothing to write, but select woke up - that means that
                        // the timeout was triggered, send some NOOP to keep the connection alive
                        queueCommand(new FSAPICommand("eval", "1"));
                    }
                    continue;
                }

                // we can read - then we do it
                if (key.isReadable()) {
                    int n = sock.read(readbuf);
                    if (n == -1) {
                        throw new IOException("end of stream");
                    }
                    if (n > 0) {
                        scanReadBuffer();
                    }
                }

                // we have something to write and we can write - then we do it
                if (writebuf.position() > 0 && key.isWritable()) {
                    synchronized (writebuf) {
                        writebuf.flip();
                        sock.write(writebuf);
                        writebuf.compact();
                    }
                }

            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                try {
                    disconnect();
                    Thread.sleep(1000);
                } catch (Exception ex) {
                }
            }
        }
    }

    /**
     * Get an attached conference by its name
     * 
     * @param name
     *            the name of the conference
     * @return the conference with the given name if it exists, null otherwise
     */
    public FSConferenceHandler getConference(String name) {
        return conferences.get(name);
    }
 
    /**
     * Registers the conference with this FSController, to pass events to it
     * 
     * @param name
     *            the conference name as used by FreeSwitch to identify the conference
     * @param conference
     *            the conference object
     */
    public void registerConference(String name, FSConferenceHandler conference) {
        conferences.put(name, conference);
    }
   
    /**
     * Unregisters the conference with this FSController
     * 
     * @param name
     *            the conference name as used by FreeSwitch to identify the conference
     */
    public void unregisterConference(String name) {
        conferences.remove(name);
    }
  
    /**
     * Registers an event callback handler with this controller
     * 
     * @param callback
     *            the callback object whose handleEvent() method will be called
     * @return 
     */
    public void addEventHandler(FSEventCallback callback) {
        synchronized (eventCallbacks) {
            eventCallbacks.add(callback);
        }
    }

    /**
     * Unregister an event callback handler
     * 
     * @param callback
     *            the callback object to unregister
     */
    public void removeEventHandler(FSEventCallback callback) {
        synchronized (eventCallbacks) {
            eventCallbacks.remove(callback);
        }
    }

    /**
     * The method originates a new FSChannel (i.e., makes an outgoing call)
     * 
     * @param aNumber the calling phone number
     * @param bNumber the called phone number      
     * @param channel the channel object that will handle the call's events
     */
    public void originateChannel(final String gateway, final FSChannel channel) {
        // remember the channel so that it gets all relevant events
        channels.put(channel.getUUID(), channel);
        queueCommand(new FSAPICommand(
                "originate",
                "{origination_uuid=" 
                    + channel.getUUID()
                    + ",return_ring_ready=true,failure_causes=user_busy,fail_on_single_reject=true,ignore_early_media=true,origination_caller_id_number="
                    + channel.getFrom() 
                    + "}"
                    + gateway
                    + "/"
                    + channel.getTo()
                    + " &park",
                new FSAPICommandCallback() {
                    
                    public void onSuccess(FSEvent event) {
                        logger.trace("originate command completed successfully");
                    }
                    
                    public void onError(FSEvent event) {
                        if (channel.getState() == FSChannel.State.INIT) {
                            logger.warn("originate command failed and channel not ACTIVE on channel {}", channel);
                            channel.setState(FSChannel.State.DEAD);
                            channels.remove(channel);                            
                        } else {
                            logger.trace("originate failed on channel {}", channel);
                        }
                        channel.onFailure();
                    }
                    
                    public void onFailure(String reply) {
                        logger.trace("originate command failed with reply {}", reply);
                    }
                    
                }));
    }

    /**
     * 
     * @param uuid
     * @return the channel mapped to the given uuid
     */
    public FSChannel getChannel(UUID uuid) {
        return channels.get(uuid);
    }
}
